package com.study.flink.datastream

import java.text.SimpleDateFormat

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * EventTime And Watermark
  *
  * 同时满足了以下2个条件，window才会触发
  * 1、watermark时间 >= window_end_time
  * 2、在[window_start_time,window_end_time)区间中有数据存在，注意是左闭右开的区间
  *
  * 1）对于out-of-order的数据，Flink可以通过watermark机制结合window的操作，来处理一定范围内的乱序数据。
  * 2）对于“迟到(late element)”太多的数据，默认是丢弃，另外还可以再提供一个宽容的时间（allowedLateness），
  * 在指定延迟时间内到达的数据还是可以触发window执行，并且是每条数据都触发了window执行。
  *
  * 对于迟到的数据，可以通过sideOutputLateData可以把迟到的数据统一收集，统计存储，方便后期排查问题。
  *
  * @author: stephen.shen
  * @create: 2019-05-31 9:17
  */
object FlinkEventTimeAndWatermarkAndApplyDemo {

  def main(args: Array[String]): Unit = {
    // 获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置使用EventTime，默认使用的是ProcessTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 设置并行度，默认为CPU核数
    env.setParallelism(1)

    // 连接socket获取输入数据
    val data: DataStream[String] = env.socketTextStream("192.168.11.11", 9999)
    // 导入隐式转换
    import org.apache.flink.api.scala._
    // 解析数据，格式：0001,1559266588844
    data.map(line => {
      val arr = line.split(",")
      (arr(0), arr(1).toLong)
    })
      // 采用周期性的触发watermark的生成和发送
      .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[(String, Long)] {
      // 当前的最大时间戳
      var currentMaxTimestamp: Long = _
      // 最大允许的乱序时间是10s
      val maxOutOfOrderness = 10000L

      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

      /**
        * 抽取数据中的时间戳
        *
        * @param element
        * @param previousElementTimestamp
        * @return event time
        */
      override def extractTimestamp(element: (String, Long), previousElementTimestamp: Long): Long = {
        val timestamp = element._2
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
        println(s"key:${element._1}, eventtime:[${element._2}|${sdf.format(element._2)}]," +
          s" currentMaxTimestamp:[${currentMaxTimestamp}|${sdf.format(currentMaxTimestamp)}]," +
          s" watermark:[${getCurrentWatermark.getTimestamp}|${sdf.format(getCurrentWatermark.getTimestamp)}]")
        timestamp
      }

      /**
        * 定义生成watermark的逻辑
        * 默认100ms被调用一次
        *
        * @return watermark
        */
      override def getCurrentWatermark: Watermark = {
        new Watermark(currentMaxTimestamp - maxOutOfOrderness)
      }
    })
      // 分组聚合
      .keyBy(0)
      // 滚动时间窗口
      .timeWindow(Time.seconds(3))
      // 允许迟到的时间
      .allowedLateness(Time.seconds(5))
      // 时间窗口内的所有数据都在这里，基于时间窗口的聚合
      .apply(new WindowFunction[(String, Long), String, Tuple, TimeWindow] {
      /**
        * 对window内的数据进行排序，保证数据的顺序
        *
        * @param key
        * @param window
        * @param input
        * @param out
        */
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[(String, Long)], out: Collector[String]): Unit = {
        val k = key.toString
        var buffer = new ListBuffer[Long]()
        val it = input.iterator
        while (it.hasNext) {
          val next = it.next()
          buffer.append(next._2)
        }
        // 排序
        buffer = buffer.sorted
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
        val result = s"key:${key}, ${buffer.size}, ${sdf.format(buffer.head)}, ${sdf.format(buffer.last)}," +
          s" window.start:${sdf.format(window.getStart)}, window.end:${sdf.format(window.getEnd)}"
        out.collect(result)
      }
    })
      // 结果打印出来
      .print()
    // 触发任务执行
    env.execute("Flink EventTime And Watermark And Apply Demo")
  }
}
